package com.to8to.kitt.esb;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;

import java.io.File;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.details.JsonInstanceSerializer;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;






import com.to8to.kitt.utils.LogUtils;
import com.to8to.kitt.utils.NettyUtils;

public abstract class AbstractDeamon{

	public static Logger logger = LoggerFactory.getLogger(AbstractDeamon.class);  
	
	protected boolean syncConfigFromZK = true;
	
	protected boolean reportAddressToZK = true;
	
	protected NodeCache privateConfigNodeCache;
	
	protected Map<String, NodeChangeEvent> privateConfigEventMap = new HashMap<String, NodeChangeEvent>();
	
	protected List<String> commonConfigPaths = new ArrayList<String>();
	
	protected List<NodeCache> commonConfigNodeCaches = new ArrayList<NodeCache>();
	
	protected ServiceDiscovery<String> serviceDiscovery;
	
	protected ServiceInstance<String> serviceInstance;
	
	protected String service;
	
	protected String instance;
	
	protected String payload;
	
	protected String reportHost;
	
	protected int reportPort = 0;
	
	protected int reportScanIntervalMS = 1000;
	
	protected boolean enableReport = true;
	


	public boolean syncConfigFromZK() {
		return syncConfigFromZK;
	}

	public void setSyncConfigFromZK(boolean syncConfigFromZK) {
		this.syncConfigFromZK = syncConfigFromZK;
	}

	public boolean reportAddressToZK() {
		return reportAddressToZK;
	}

	public void setReportAddressToZK(boolean reportAddressToZK) {
		this.reportAddressToZK = reportAddressToZK;
	}


	public int getReportScanIntervalMS() {
		return reportScanIntervalMS;
	}

	public void setReportScanIntervalMS(int reportScanIntervalMS) {
		this.reportScanIntervalMS = reportScanIntervalMS;
	}

	public String getReportHost() {
		return reportHost;
	}
	
	public int getReportPort() {
		return reportPort;
	}

	public void setReportPort(int reportPort) {
		this.reportPort = reportPort;
	}

	public void setReportHost(String reportHost) {
		this.reportHost = reportHost;
	}

	public void addCommonConfigPath(String path)
	{
		this.commonConfigPaths.add(path);
	}
	
	public void removeCommonConfigPath(String path)
	{
		this.commonConfigPaths.remove(path);
	}

	public String privateConfigPath()
	{
		if(service ==  null)
			return null;
		
		if(instance == null)
			return null;
		
		return ZK.ZK_CONFIG.configPath + "/" + service + "/" + instance;
	}
	
	public String commonConfigPath(String relativePath)
	{
		return ZK.ZK_CONFIG.configPath + "/" + relativePath ;
	}
	
	
	public void setService(String service)
	{
		this.service = service;
	}
	
	public void setInstance(String instance)
	{
		this.instance = instance;
	}
	
	public void setServiceInstance(String service, String instance)
	{
		this.service = service;
		this.instance = instance;
	}
	
	public String getServiceInstance()
	{
		return service + "/" + instance;
	}
	
	protected boolean isAvailable() 
	{
		return true;
	}
	
	protected void report(boolean unreg)
	{
		logger.info("<<<< report my address to ZK .... start! >>>>");
		try {
			
			if(serviceInstance != null && unreg)
				serviceDiscovery.unregisterService(serviceInstance);
			
//			if(serviceInstance == null || newinstance)
				serviceInstance = ServiceInstance.<String>builder().id(instance).name(service).address(reportHost).payload(payload).port(reportPort).build();
			
			String dir = System.getProperty("java.run.dir");
			
			String prog = System.getProperty("java.apps.prog");
			
			if(!StringUtils.isEmpty(dir) && !StringUtils.isEmpty(prog))
			{
				File file = new File(dir + "/" + prog + ".zk");
				String pid = NettyUtils.getPID();
				FileUtils.writeStringToFile(file, pid + "\n" +  ZK.ZK_CONFIG.servicePath + "/" + service + "/" + instance + "\n");
			}
			
			serviceDiscovery.registerService(serviceInstance);
			
//			if(!reportScanThread.isAlive())
//				reportScanThread.start();
			
			logger.info("<<<< report my address to ZK .... success! >>>>");
			
		} catch (Exception e) {
			e.printStackTrace();
			logger.error(LogUtils.logStackTrace(e));
			logger.info("############# report my address to ZK .... failed! ##############");
		}
	}
	
	
	

	
	protected boolean initZkConfigListener()
	{	
		logger.info("<<<<< initZkConfigListener start! >>>>>");

		logger.info("<<<<< init listeners to common config paths ... start! >>>>>");
		
		for(String commonConfigPath : commonConfigPaths)
		{
			try {
				
				String abspath = commonConfigPath(commonConfigPath);
				
				logger.info(".....init common config path:{}",abspath );
				
				byte[] data = ZK.ZK_CLIENT.getData().forPath(abspath);
				
				NodeChangeEvent event = new NodeChangeEvent();
				event.odata = null;
				event.ndata = data;
				event.path = abspath;
				
				this.onCommonConfigChange(event);
				
				event.odata = event.ndata;
				
				NodeCache cache = new NodeCache(ZK.ZK_CLIENT, abspath);
				
				cache.start();
				
				cache.getListenable().addListener(new ZKNodeCacheListener(ZK.ZK_CLIENT, abspath){

					@Override
					public void onNodeChanged(NodeChangeEvent e) throws Exception {
						AbstractDeamon.this.onCommonConfigChange(e);
						e.odata = e.ndata;
					}});
		        
				this.commonConfigNodeCaches.add(cache);
				
			} catch (Throwable e) {
				
				NettyUtils.exitIfOutOfMemory(e);
				
				logger.error(LogUtils.logStackTrace(e));
			}
		}
		
		logger.info("<<<<< init listeners to common config paths ... finish! >>>>>");
		
		String path = privateConfigPath();
		
		logger.info("<<<<< init listeners to private config path:{} ... start! >>>>>", path);
		
		try {
			
			if(path == null)
			{
				logger.error("zkconfig path is null!");
				return false;
			}
			
			Stat stat = ZK.ZK_CLIENT.checkExists().forPath(path);
			
			if(stat == null)
			{
				logger.error("node not exist: " + path);
				return false;
			}
			
			List<String> children = ZK.ZK_CLIENT.getChildren().forPath(path);
			
			for(String child: children)
			{
				byte[] data = ZK.ZK_CLIENT.getData().forPath(path + "/" + child);
				NodeChangeEvent e = new NodeChangeEvent();
				e.path = path + "/" + child;
				e.odata = null;
				e.ndata = data;
				privateConfigEventMap.put(child, e);
			}
			
			this.onPrivateConfigChange(privateConfigEventMap);
			
			for(NodeChangeEvent event: privateConfigEventMap.values())
				event.odata = event.ndata;
			
			privateConfigNodeCache = new NodeCache(ZK.ZK_CLIENT, path);
			
			privateConfigNodeCache.start();
			
	        privateConfigNodeCache.getListenable().addListener(new ZKNodeCacheListener(ZK.ZK_CLIENT, path){

				@Override
				public void onNodeChanged(NodeChangeEvent e) throws Exception {

					String path = AbstractDeamon.this.privateConfigPath();
					
					List<String> children = ZK.ZK_CLIENT.getChildren().forPath(path);
					
					for(String child: children)
					{
						byte[] data = ZK.ZK_CLIENT.getData().forPath(path + "/" + child);
						NodeChangeEvent event = AbstractDeamon.this.privateConfigEventMap.get(child);
						if(event != null)
							event.ndata = data;
					}
					
					AbstractDeamon.this.onPrivateConfigChange(privateConfigEventMap);
					
					for(NodeChangeEvent ee : AbstractDeamon.this.privateConfigEventMap.values())
						ee.odata = ee.ndata;
					
				}});
	        

		} catch (Exception e) {
			logger.error(LogUtils.logStackTrace(e));
		}
		
		logger.info("<<<<< init listeners to private config path ... finish! >>>>>");
		
		logger.info("<<<<< initZkConfigListener finish! >>>>>");
		
		return true;
	}
	
	protected void reloadLogback(String xml)
	{
		String real = xml.replace("${java.service}", service).replace("${java.instance}", instance);
		System.out.println("reloadLogback:\n " + real);
		LogUtils.resetConfigByData(real);
	}
	
	protected boolean initZkServiceReporter()
	{
		logger.info("<<<<< initZkServiceReporter: \n" + ZK.ZK_CONFIG.servicePath + " >>>>>");
		
		serviceDiscovery = ServiceDiscoveryBuilder.builder(String.class)
				.client(ZK.ZK_CLIENT)
				.serializer(new JsonInstanceSerializer<String>(String.class))
				.basePath(ZK.ZK_CONFIG.servicePath).build();
		
		try {
			
			serviceDiscovery.start();
			
			logger.info("<<<<< initZkServiceReporter success! >>>>>");
			
			return true;
			
		} catch (Exception e) {
			logger.error(LogUtils.logStackTrace(e));
		}
		
		logger.error("<<<<< initZkServiceReporter failed! >>>>>");
		
		return false;
	}
	
	
	public boolean init()
	{
		logger.info(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> start initializing ...");
		
		if(syncConfigFromZK || reportAddressToZK)
		{
			if(service == null || instance == null)
			{
				logger.error("please set service and instance before listen to zk!");
				return false;
			}
			
			if(reportAddressToZK)
				if(!initZkServiceReporter())
					return false;
			
			if(syncConfigFromZK)
				if(!initZkConfigListener())
					return false;
					
		}
		
		System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
		System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
		System.out.println("@@@@@@@@@@@@@@@@@@@@@      deamon process init success!!!        @@@@@@@@@@@@@@@@@@@");
		System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
		System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
		
		
		
		return true;
	}
	
	abstract public void onPrivateConfigChange(Map<String, NodeChangeEvent> config);
	
	abstract public void onCommonConfigChange(NodeChangeEvent e);
}
